package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.function.KeyFunction;
import cn.itcast.function.StockIncrWindowFunction;
import cn.itcast.inter.ProcessDataInterface;
import cn.itcast.util.DorisUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * 涨跌幅分时行情 ->Druid
 * 1.每分钟生成最新时间的一条数据
 */
public class StockRiseAndFallTask implements ProcessDataInterface {
    @Override
    public void process(DataStream<CleanBean> waterData) {

        /**
         * 1.数据分组
         * 2.划分时间窗口
         * 3.创建bean对象
         * 4.分时数据处理（新建分时窗口函数）
         * 5.数据存储(单表)
         */
        //1.数据分组
        waterData.keyBy(new KeyFunction())
                .window(TumblingEventTimeWindows.of(Time.seconds(60)))
                .apply(new StockIncrWindowFunction())
                .addSink(DorisUtil.getSinkFunction("insert into quot_stock_up_down values(?,?,?,?,?,?,?,?,?,?,?,?,?)"));

    }
}
